package com.shujia.flink.soure

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo2SourceFunaction {
  def main(args: Array[String]): Unit = {

    //创建flink的运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //使用自定义source
    val myDS: DataStream[Int] = env.addSource(new MySourceFunaction)

    myDS.print()

    env.execute()


  }
}

class MySourceFunaction extends SourceFunction[Int] {
  /**
    * 在方法中生成数据
    *
    * @param ctx
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    /*var l = 0 to 10

    有界流
    for (i <- l) {

      //将数据发送到下游
      ctx.collect(i)

    }*/

    var i = 0

    while (true) {

      ctx.collect(i)
      i += 1
    }

  }

  /**
    * 任务结束之后会调用的方法，一般用于回收资源
    *
    */

  override def cancel(): Unit = {

  }
}
